/*
 * @(#)OfflineLifeCycleServer.java   1.0  2021年11月17日
 * 
 * Copyright (c)	2014-2020. All Rights Reserved.	GuangZhou hhmk Technology Company LTD.
 */
package com.swift.dts.offline;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.swift.dts.dao.OfflineStateInfoDao;
import com.swift.dts.dao.TablesDao;
import com.swift.dts.dao.model.OfflineState;
import com.swift.dts.dao.model.OfflineStateInfo;
import com.swift.dts.dao.model.SynObjectKey;
import com.swift.dts.model.Table;
import com.swift.dts.offline.exec.CheckAndCreateSynTable;
import com.swift.dts.offline.exec.ExecOfflineSql;
import com.swift.dts.offline.exec.SynOfflineData;
import com.swift.dts.util.CreateExecSql;
import com.swift.util.date.DateUtil;
import com.swift.util.type.TypeUtil;

/**
 * 添加说明 
 * @author zhengjiajin
 * @version 1.0 2021年11月17日
 */
@Component
public class OfflineLifeCycleServer implements OfflineLifeCycle{

    private static final Logger log = LoggerFactory.getLogger(OfflineLifeCycleServer.class);
   
    @Autowired
    private OfflineStateInfoDao offlineStateInfoDao;
    
    @Autowired
    private CheckAndCreateSynTable checkAndCreateSynTable;
    
    @Autowired
    private SynOfflineData synOfflineData;
    
    @Autowired
    private TablesDao tablesDao;
    
    @Autowired
    private ExecOfflineSql execOfflineSql;
    //同时同步15个表
    private ExecutorService executorService = Executors.newFixedThreadPool(15);
    
    private Map<SynObjectKey,ThreadOfflineSyn> onRunOfflineSyn = new ConcurrentHashMap<>();
    /** 
     * @see com.swift.dts.offline.OfflineLifeCycle#start(com.swift.dts.dao.model.SynObjectKey)
     */
    @Override
    public synchronized void start(SynObjectKey key) {
        if(onRunOffline(key)) return;
        ThreadOfflineSyn thread= new ThreadOfflineSyn(key);
        onRunOfflineSyn.put(key, thread);
        OfflineStateInfo info = offlineStateInfoDao.get(key);
        if(info==null) {
            info = new OfflineStateInfo();
            info.setCreateTime(DateUtil.formatDate(new Date()));
            info.setDataSource(key.getDataSource());
            info.setDbName(key.getDbName());
            info.setDestination(key.getDestination());
            info.setPercent(0d);
            info.setState(OfflineState.NO_START.getState());
            info.setTableName(key.getTableName());
            offlineStateInfoDao.save(info);
        }else {
            offlineStateInfoDao.updateState(key, OfflineState.NO_START, "");
        }
        executorService.execute(thread);
    }
    
    private class ThreadOfflineSyn implements Runnable {
        
        private AtomicBoolean start = new AtomicBoolean(true);
        
        private SynObjectKey key;
        
        private ThreadOfflineSyn(SynObjectKey key) {
            this.key=key;
        }

        /** 
         * @see java.lang.Runnable#run()
         */
        @Override
        public void run() {
            offlineStateInfoDao.updateState(key, OfflineState.STARTED,null);
            try {
                //初始化表
                if(start.get()) checkAndCreateSynTable.checkAndCreateTable(key);
                //同步离线数据
                if(start.get()) synOfflineData.synOfflineData(key);
                //执行离线期间的BINLOG
                if(start.get()) execOfflineSql.execOfflineSql(key);
                if(start.get()) offlineStateInfoDao.updateState(key, OfflineState.SECCESS,null);
                onRunOfflineSyn.remove(key);
            }catch(Throwable ex) {
                try {
                    log.error("执行离线任务异常:",ex);
                    onRunOfflineSyn.remove(key);
                    String errorRem = ex.getMessage();
                    if(errorRem.length()>500) errorRem=errorRem.substring(0,500);
                    offlineStateInfoDao.updateState(key, OfflineState.ERROR ,CreateExecSql.formatValue(errorRem));
                }catch(Throwable e) {
                    log.error("提交修改异常:",e);
                }
            }
        }
        
        private void stop() {
            start.set(false);
            synOfflineData.stop(key);
            execOfflineSql.cleanOfflineSql(key);
            offlineStateInfoDao.updateState(key, OfflineState.DEL, "");
        }

    }

    /** 
     * @see com.swift.dts.offline.OfflineLifeCycle#offlineState(com.swift.dts.dao.model.SynObjectKey)
     */
    @Override
    public boolean onRunOffline(SynObjectKey key) {
        return onRunOfflineSyn.containsKey(key);
    }
    
    /** 
     * @see com.swift.dts.offline.OfflineLifeCycle#delete(com.swift.dts.dao.model.SynObjectKey)
     */
    @Override
    public void delete(SynObjectKey key) {
        ThreadOfflineSyn syn = onRunOfflineSyn.remove(key);
        if(syn!=null) syn.stop();
    }
    
    @Override
    public void startOfflineTash(String dataSource,String destination) {
        List<Table> offlineTable = getStartOfflineTable(dataSource,destination);
        if(TypeUtil.isNull(offlineTable)) return;
        //开启离线同步任务
        for(Table dbTable:offlineTable) {
            SynObjectKey key = createKey(dataSource, destination, dbTable);
            this.start(key);
        }
    }
    
    private SynObjectKey createKey(String dataSource,String destination,Table table) {
        SynObjectKey key = new SynObjectKey();
        key.setDataSource(dataSource);
        key.setDbName(table.getDbName());
        key.setDestination(destination);
        key.setTableName(table.getTableName());
        return key;
    }
    
    //找出所有需要启动离线任务的表
    private List<Table> getStartOfflineTable(String dataSource,String destination) {
        //找出所有需要同步的表
        List<Table> synTables = tablesDao.getAllSynTables(dataSource);
        if(TypeUtil.isNull(synTables)) return null;
        //过滤掉已经开始离线同步的表
        List<OfflineStateInfo> offList = offlineStateInfoDao.list();
        return synTables.stream().filter(tab->checkNotInOffline(dataSource,destination,tab, offList)).collect(Collectors.toList());
    }
    
    private boolean checkNotInOffline(String dataSource,String destination,Table table,List<OfflineStateInfo> offList) {
        if(onRunOffline(createKey(dataSource, destination, table))) return false;//正在执行的不执行
        if(TypeUtil.isNull(offList)) return true;
        for(OfflineStateInfo info:offList) {
            if(!info.getDbName().equalsIgnoreCase(table.getDbName())) continue;
            if(!info.getTableName().equalsIgnoreCase(table.getTableName())) continue;
            if(!info.getDataSource().equalsIgnoreCase(dataSource)) continue;
            if(!info.getDestination().equalsIgnoreCase(destination)) continue;
            ////0未开始1执行中2已完成3执行异常4删除同步
            if(info.getState()==1) return false;//1执行中
            if(info.getState()==2) return false;//执行完的不执行
            if(info.getState()==3) return false;//3执行异常
        }
        return true;
    }

}
